package org.budo.dubbo.protocol.async.repository.redis;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import org.budo.dubbo.protocol.async.BudoAsyncDubboProtocol;
import org.budo.dubbo.protocol.async.repository.AbstractAsyncRepository;
import org.budo.dubbo.protocol.async.repository.BudoAsyncInvocation;
import org.budo.dubbo.protocol.async.util.AsyncSerializeUtil;
import org.budo.dubbo.protocol.async.util.DestinationUtil;
import org.budo.dubbo.protocol.async.util.RepositoryUtil;
import org.budo.support.lang.util.StringUtil;

import com.alibaba.dubbo.rpc.Exporter;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.RpcInvocation;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPubSub;

/**
 * @author limingwei
 */
@Slf4j
@Getter
@Setter
public class RedisAsyncRepository extends AbstractAsyncRepository {
    private JedisPool jedisPool;

    @Override
    public Integer popMessage(Exporter<?> exporter) {
        String[] channels = this.channels();

        if (null == channels || channels.length < 1) {
            log.error("#55 channels=" + channels + ", this=" + this);
            return 0;
        }

        JedisPubSub jedisPubSub = new JedisPubSub() {
            public void onMessage(String channel, String message) {
                Invoker invoker = RedisAsyncRepository.this.getInvoker(channel);

                BudoAsyncInvocation budoAsyncInvocation = AsyncSerializeUtil.deserializeAsyncInvocation(message.getBytes(), invoker);

                RpcInvocation invocation = BudoAsyncInvocation.toDubboInvocation(budoAsyncInvocation);
                invoker.invoke(invocation); // 执行方法
            }
        };

        Jedis jedis = this.getJedisPool().getResource();
        jedis.subscribe(jedisPubSub, channels);
        return -1;
    }

    @Override
    public Integer popMessageAndInvoke(Invoker invoker) {
        throw new RuntimeException();
    }

    @Override
    public void sendMessageWhenInvoke(Invocation invocation) {
        String destinationName = DestinationUtil.getDestinationName(invocation.getInvoker());

        Jedis jedis = this.getJedisPool().getResource();

        BudoAsyncInvocation budoAsyncInvocation = BudoAsyncInvocation.fromDubboInvocation(invocation);
        byte[] requestBody = AsyncSerializeUtil.serializeAsyncInvocation(budoAsyncInvocation, invocation.getInvoker());

        jedis.publish(destinationName, new String(requestBody));
        jedis.close();
    }

    protected Invoker getInvoker(String channel) {
        BudoAsyncDubboProtocol dubboProtocol = BudoAsyncDubboProtocol.getInstance();

        if (null == dubboProtocol || null == dubboProtocol.getExporterMap() || dubboProtocol.getExporterMap().isEmpty()) {
            if (log.isDebugEnabled()) {
                log.debug("#154 dubboPopMessage error, protocol=" + dubboProtocol + ", this=" + this);
            }

            return null;
        }

        Collection<Exporter<?>> dubboExporters = dubboProtocol.getExporterMap().values();
        for (Exporter<?> dubboExporter : dubboExporters) { // 每一个
            Invoker<?> invoker = dubboExporter.getInvoker();
            String destinationName = DestinationUtil.getDestinationName(invoker);

            if (StringUtil.equals(destinationName, channel)) {
                return invoker;
            }
        }

        log.error("#105 channel=" + channel + ", invoker is null");
        return null;
    }

    private String[] channels() {
        BudoAsyncDubboProtocol dubboProtocol = BudoAsyncDubboProtocol.getInstance();

        if (null == dubboProtocol || null == dubboProtocol.getExporterMap() || dubboProtocol.getExporterMap().isEmpty()) {
            if (log.isDebugEnabled()) {
                log.debug("#154 dubboPopMessage error, protocol=" + dubboProtocol + ", this=" + this);
            }

            return null;
        }

        List<String> destinations = new ArrayList<String>();

        Collection<Exporter<?>> dubboExporters = dubboProtocol.getExporterMap().values();
        for (Exporter<?> dubboExporter : dubboExporters) { // 每一个
            Invoker<?> invoker = dubboExporter.getInvoker();
            String asyncRepositoryBeanName = RepositoryUtil.getAsyncRepositoryBeanName(invoker);

            if (!StringUtil.equals(asyncRepositoryBeanName, this.getBeanName())) { // 只处理指定当前Bean的
                if (log.isDebugEnabled()) {
                    log.debug("#175 asyncRepositoryBeanName=" + asyncRepositoryBeanName + ", this=" + this);
                }

                continue;
            }

            String destinationName = DestinationUtil.getDestinationName(invoker);
            destinations.add(destinationName);
        }

        return destinations.toArray(new String[destinations.size()]);
    }

    @Override
    public void sendMessage(String destinationName, Object message) {
        throw new RuntimeException();
    }

    @Override
    public void sendRawMessage(String destinationName, Object rawMessage) {
        throw new RuntimeException();
    }

    @Override
    public void destroy() throws Exception {
    }
}